package com.nokia.analytics.avrotest.AvroPositioningETL;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;

import org.apache.avro.mapred.AvroKey;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.log4j.Logger;

import com.nokia.analytics.AvroTypes.Position;

public class AvroPositioningBinaryMapper extends
		Mapper<LongWritable, Text, AvroKey<Position>, NullWritable> {
	
	private AvroKey<Position> pStats;
	public static final Logger LOGGER = Logger
			.getLogger(AvroPositioningBinaryMapper.class.getName());
	
	private static final int HEADER_LENGTH = 14;
	private static final int START_HEADER_LENGTH = 4;

	@Override
	protected void setup(Context context) {
		// mStats = new AvroKey<GenericData.Record>(null);
		pStats = new AvroKey<Position>();
	}

	@Override
	protected void map(LongWritable fileByteOffset, Text value, Context context)
			throws IOException, InterruptedException {

		Position pStatsDatum = new Position();

		DecodeBase64 db64 = new DecodeBase64(value.toString());
		byte[] decoded = db64.getBytes();
		
		// Check that we have a least as many bytes as header requires.
		if (decoded.length < START_HEADER_LENGTH) {
			LOGGER.error("Base64 input row does not have enough bytes to read the size value at offset: "
					+ (START_HEADER_LENGTH) + " Got: " + decoded.length);
			return;
		}

		InputStream is = new ByteArrayInputStream(decoded);
		DataInputStream dis = new DataInputStream(is);

		// header
		byte major_version = (byte) dis.readUnsignedByte();
		int major_version_unsigned = BinUtil.convertToUnsignedByte(major_version);

		byte minor_version = (byte) dis.readUnsignedByte();
		int minor_version_unsigned = BinUtil.convertToUnsignedByte(minor_version);

		int size = BinUtil.convertToUnsignedShort(BinUtil.L2BEndianShort(dis
				.readUnsignedShort()));
		
		// size validation
		if (size != (decoded.length - HEADER_LENGTH)) {
			LOGGER.error("Size of payload was wrong, expected: "
					+ (decoded.length - HEADER_LENGTH) + " Got: " + size);
			return;
		}

		//short checksum = L2BEndianShort(dis.readUnsignedShort());
		// Get past checksum which is not used.
		BinUtil.L2BEndianShort(dis.readUnsignedShort());

		int validity_mask = BinUtil.L2BEndianInt(dis.readInt());
		int presence_mask = BinUtil.L2BEndianInt(dis.readInt());

		// Mandatory payload
		// Expect these values to always be present, valid or not.
		long coordinates_timestamp = BinUtil.L2BEndianLong(dis.readLong());

		int latitude = BinUtil.L2BEndianInt(dis.readInt());

		int longitude = BinUtil.L2BEndianInt(dis.readInt());

		short altitude = BinUtil.L2BEndianShort(dis.readUnsignedShort());

		int haccuracy = BinUtil.L2BEndianInt(dis.readInt());
		long haccuracy_unsigned = BinUtil.convertToUnsignedInt(haccuracy);

		short vaccuracy = BinUtil.L2BEndianShort(dis.readUnsignedShort());
		int vaccuracy_unsigned = BinUtil.convertToUnsignedShort(vaccuracy);

		byte network_type = (byte) dis.readUnsignedByte();
		int network_type_unsigned = BinUtil.convertToUnsignedByte(network_type);

		short serving_mcc_systemID = BinUtil.L2BEndianShort(dis.readUnsignedShort());
		int serving_mcc_systemID_unsigned = BinUtil.convertToUnsignedShort(serving_mcc_systemID);

		short serving_mnc_networkID = BinUtil.L2BEndianShort(dis.readUnsignedShort());
		int serving_mnc_networkID_unsigned = BinUtil.convertToUnsignedShort(serving_mnc_networkID);

		short home_mcc_systemID = BinUtil.L2BEndianShort(dis.readUnsignedShort());
		int home_mcc_systemID_unsigned = BinUtil.convertToUnsignedShort(home_mcc_systemID);

		short home_mnc_networkID = BinUtil.L2BEndianShort(dis.readUnsignedShort());
		int home_mnc_networkID_unsigned = BinUtil.convertToUnsignedShort(home_mnc_networkID);

		byte provided_measurements = (byte) dis.readUnsignedByte();
		int provided_measurements_unsigned = BinUtil.convertToUnsignedByte(provided_measurements);

		byte used_measurements = (byte) dis.readUnsignedByte();
		int used_measurements_unsigned = BinUtil.convertToUnsignedByte(used_measurements);

		byte protocol_type = (byte) dis.readUnsignedByte();
		int protocol_type_unsigned = BinUtil.convertToUnsignedByte(protocol_type);

		Short protocol_version = BinUtil.L2BEndianShort(dis.readUnsignedShort());
		int protocol_version_unsigned = BinUtil.convertToUnsignedShort(protocol_version);
		Integer protocol_version_major = protocol_version_unsigned >> 8;
		Integer protocol_version_minor = protocol_version_unsigned & 0x000000ff;
		int protocol_version_major_unsigned = BinUtil.convertToUnsignedByte(protocol_version_major
				.byteValue());
		int protocol_version_minor_unsigned = BinUtil.convertToUnsignedByte(protocol_version_minor
				.byteValue());


		byte manufacturer = (byte) dis.readUnsignedByte();
		int manufacturer_unsigned = BinUtil.convertToUnsignedByte(manufacturer);

		byte client_type = (byte) dis.readUnsignedByte();
		int client_type_unsigned = BinUtil.convertToUnsignedByte(client_type);

		// status is derived from validity mask.
		boolean status = ((validity_mask & (int) Math.pow(2, 17)) == Math.pow(
				2, 17));

		// Optional fields
		// check presence bit mask to determine which to read.
		Short error_code = null;
		if ((presence_mask & (int) Math.pow(2, 0)) == Math.pow(2, 0)) {
			error_code = BinUtil.L2BEndianShort(dis.readUnsignedShort());
		}

		// utfstring modelID
		String model_id = null;
		if ((presence_mask & (int) Math.pow(2, 1)) == Math.pow(2, 1)) {
			Integer model_id_length = dis.readUnsignedByte();

			byte[] model_id_bytes = new byte[model_id_length];
			dis.readFully(model_id_bytes, 0, model_id_length);
			model_id = new String(model_id_bytes, "UTF-8");
		}

		// utfstring client name
		String client_name = null;
		if ((presence_mask & (int) Math.pow(2, 2)) == Math.pow(2, 2)) {
			Integer client_name_length = dis.readUnsignedByte();

			byte[] client_name_bytes = new byte[client_name_length];
			dis.readFully(client_name_bytes, 0, client_name_length);
			client_name = new String(client_name_bytes, "UTF-8");
		}

		// utfstring fqdn
		String fqdn = null;
		if ((presence_mask & (int) Math.pow(2, 3)) == Math.pow(2, 3)) {
			Integer fqdn_length = dis.readUnsignedByte();

			byte[] fqdn_bytes = new byte[fqdn_length];
			dis.readFully(fqdn_bytes, 0, fqdn_length);
			fqdn = new String(fqdn_bytes, "UTF-8");
		}

		Byte sim2_network_type = null;
		Integer sim2_network_type_unsigned = null;
		if ((presence_mask & (int) Math.pow(2, 4)) == Math.pow(2, 4)) {
			sim2_network_type = (byte) dis.readUnsignedByte();
			sim2_network_type_unsigned = BinUtil.convertToUnsignedByte(sim2_network_type);
		}

		Short sim2_serving_mcc_systemID = null;
		Integer sim2_serving_mcc_systemID_unsigned = null;
		if ((presence_mask & (int) Math.pow(2, 5)) == Math.pow(2, 5)) {
			sim2_serving_mcc_systemID = (short) BinUtil.L2BEndianShort(dis
					.readUnsignedShort());
			sim2_serving_mcc_systemID_unsigned = BinUtil.convertToUnsignedShort(sim2_serving_mcc_systemID);
		}

		Short sim2_serving_mnc_networkID = null;
		Integer sim2_serving_mnc_networkID_unsigned = null;
		if ((presence_mask & (int) Math.pow(2, 6)) == Math.pow(2, 6)) {
			sim2_serving_mnc_networkID = (short) BinUtil.L2BEndianShort(dis
					.readUnsignedShort());
			sim2_serving_mnc_networkID_unsigned = BinUtil.convertToUnsignedShort(sim2_serving_mnc_networkID);
		}

		Short sim2_home_mcc_systemID = null;
		Integer sim2_home_mcc_systemID_unsigned = null;
		if ((presence_mask & (int) Math.pow(2, 7)) == Math.pow(2, 7)) {
			sim2_home_mcc_systemID = (short) BinUtil.L2BEndianShort(dis
					.readUnsignedShort());
			sim2_home_mcc_systemID_unsigned = BinUtil.convertToUnsignedShort(sim2_home_mcc_systemID);
		}

		Short sim2_home_mnc_networkID = null;
		Integer sim2_home_mnc_networkID_unsigned = null;
		if ((presence_mask & (int) Math.pow(2, 8)) == Math.pow(2, 8)) {
			sim2_home_mnc_networkID = (short) BinUtil.L2BEndianShort(dis
					.readUnsignedShort());
			sim2_home_mnc_networkID_unsigned = BinUtil.convertToUnsignedShort(sim2_home_mnc_networkID);
		}

		Byte sim2_provided_measurements = null;
		Integer sim2_provided_measurements_unsigned = null;
		if ((presence_mask & (int) Math.pow(2, 9)) == Math.pow(2, 9)) {
			sim2_provided_measurements = (byte) dis.readUnsignedByte();
			sim2_provided_measurements_unsigned = BinUtil.convertToUnsignedByte(sim2_provided_measurements);
		}

		Byte sim2_used_measurements = null;
		Integer sim2_used_measurements_unsigned = null;
		if ((presence_mask & (int) Math.pow(2, 10)) == Math.pow(2, 10)) {
			sim2_used_measurements = (byte) dis.readUnsignedByte();
			sim2_used_measurements_unsigned = BinUtil.convertToUnsignedByte(sim2_used_measurements);
		}
		pStatsDatum.setMajorVersion(major_version_unsigned);
		pStatsDatum.setMinorVersion(minor_version_unsigned);
		pStatsDatum.setValidityMask(validity_mask);

		if ((validity_mask & (int) Math.pow(2, 0)) == Math.pow(2, 0)) {
			pStatsDatum.setCoordinatesTimestamp((double)coordinates_timestamp);
		}
		if ((validity_mask & (int) Math.pow(2, 1)) == Math.pow(2, 1)) {
			pStatsDatum.setLatitude(BinUtil.crdInterpreted(latitude));
		}
		if ((validity_mask & (int) Math.pow(2, 2)) == Math.pow(2, 2)) {
			pStatsDatum.setLongitude(BinUtil.crdInterpreted(longitude));
		}

		if ((validity_mask & (int) Math.pow(2, 3)) == Math.pow(2, 3)) {
			pStatsDatum.setAltitude((int)altitude);
		}
		if ((validity_mask & (int) Math.pow(2, 4)) == Math.pow(2, 4)) {
			pStatsDatum.setHaccuracy((double)haccuracy_unsigned);
		}
		if ((validity_mask & (int) Math.pow(2, 5)) == Math.pow(2, 5)) {
			pStatsDatum.setVaccuracy(vaccuracy_unsigned);
		}
		if ((validity_mask & (int) Math.pow(2, 6)) == Math.pow(2, 6)) {
			pStatsDatum.setNetworkType(network_type_unsigned);
		}
		if ((validity_mask & (int) Math.pow(2, 7)) == Math.pow(2, 7)) {
			pStatsDatum.setServingMccSystemID(serving_mcc_systemID_unsigned);
		}
		if ((validity_mask & (int) Math.pow(2, 8)) == Math.pow(2, 8)) {
			pStatsDatum.setServingMncNetworkID(serving_mnc_networkID_unsigned);
		}
		if ((validity_mask & (int) Math.pow(2, 9)) == Math.pow(2, 9)) {
			pStatsDatum.setHomeMccSystemID(home_mcc_systemID_unsigned);
		}
		if ((validity_mask & (int) Math.pow(2, 10)) == Math.pow(2, 10)) {
			pStatsDatum.setHomeMncNetworkID(home_mnc_networkID_unsigned);
		}
		if ((validity_mask & (int) Math.pow(2, 11)) == Math.pow(2, 11)) {
			pStatsDatum.setProvidedMeasurements(provided_measurements_unsigned);
		}
		if ((validity_mask & (int) Math.pow(2, 12)) == Math.pow(2, 12)) {
			pStatsDatum.setUsedMeasurements(used_measurements_unsigned);
		}
		if ((validity_mask & (int) Math.pow(2, 13)) == Math.pow(2, 13)) {
			pStatsDatum.setProtocolType(protocol_type_unsigned);
		}
		if ((validity_mask & (int) Math.pow(2, 14)) == Math.pow(2, 14)) {
			pStatsDatum.setProtocolVersion(protocol_version_unsigned);
			pStatsDatum.setProtocolVersionMajor(protocol_version_major_unsigned);
			pStatsDatum.setProtocolVersionMinor(protocol_version_minor_unsigned);
		}
		if ((validity_mask & (int) Math.pow(2, 15)) == Math.pow(2, 15)) {
			pStatsDatum.setManufacturer(manufacturer_unsigned);
		}
		if ((validity_mask & (int) Math.pow(2, 16)) == Math.pow(2, 16)) {
			pStatsDatum.setClientType(client_type_unsigned);
		}

		// status is the boolean value of the validity mask at position
		// 17
		// (based on counting from 0)
		pStatsDatum.setStatus(status);
		//
		// Optional payload (pre-calculated based on presence bitmask)
		if (error_code != null) {
			pStatsDatum.setErrorCode((int)error_code);
		}
		if (model_id != null) {
			pStatsDatum.setModelId(model_id);
		}
		if (client_name != null) {
			pStatsDatum.setClientName(client_name);
		}
		if (fqdn != null) {
			pStatsDatum.setFqdn(fqdn);
		}
		if (sim2_network_type != null) {
			pStatsDatum.setSim2NetworkType(sim2_network_type_unsigned);
		}
		if (sim2_serving_mcc_systemID != null) {
			pStatsDatum.setSim2ServingMccSystemID(sim2_serving_mcc_systemID_unsigned);
		}
		if (sim2_serving_mnc_networkID != null) {
			pStatsDatum.setSim2ServingMncNetworkID(sim2_serving_mnc_networkID_unsigned);
		}
		if (sim2_home_mcc_systemID != null) {
			pStatsDatum.setSim2HomeMccSystemID(sim2_home_mcc_systemID_unsigned);
		}
		if (sim2_home_mnc_networkID != null) {
			pStatsDatum.setSim2HomeMncNetworkID(sim2_home_mnc_networkID_unsigned);
		}
		if (sim2_provided_measurements != null) {
			pStatsDatum.setSim2ProvidedMeasurements(sim2_provided_measurements_unsigned);
		}
		if (sim2_used_measurements != null) {
			pStatsDatum.setSim2UsedMeasurements(sim2_used_measurements_unsigned);
		}
		pStats.datum(pStatsDatum);
		
		context.write(pStats , NullWritable.get());
	}
}
